package com.wangzhenyu.learn.test.kotlin.coroutine_

import com.wangzhenyu.common.util.log.LogUtil
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.selects.select
import java.io.Serializable
import kotlin.coroutines.*



//Coroutine的使用




//输出当前协程的名字
fun printlnThread(any: Any) {
    //注意要先设置VM值
    println("$any:\tthread:${Thread.currentThread().name}")
}

//launch
//这里主要叙述launch的流程
//首先，因为block 的类型为 suspend CoroutineScope.() -> Unit,后面的R为CoroutineScope,T为Unit
//先生成一个coroutine(两个版本，目前用StandaloneCoroutine，这是AbstractCoroutine的子类)
//调用coroutine的start方法，并传入start:CoroutineStart，receiver:R，block:suspend R.() -> T(这里的receiver传的是coroutine)
//在这个start方法中，实际上是调用 start:CoroutineStart 的invoke方法，并传入 invoke:suspend R.() -> T ,
//receiver:R , completion: Continuation<T>(这个传的其实是coroutine)
//在这里是DEFAULT分支，即调用block.startCoroutineCancellable(receiver, completion)
//这个方法中，有以下方法
//(suspend (R) -> T).createCoroutineUnintercepted
//在该方法中，block这个Lambda表达式的类型为Function2，但是对应的具体类为SuspendLambda，所以会调用create方法
//Continuation<T>.intercepted()
//该方法的主要作用就是生成DispatchedContinuation。它由CoroutineDispatcher和Continuation两部分组成。Dispatcher决定在Continuation在哪个线程中执行
//大概就是这样


//CoroutineContext
//  协程上下文
//  控制协程在哪个线程中执行
//  设置协程的名字
//  捕获协程抛出的异常
//  CoroutineContext是element的set集合，没有重复类型的element对象。集合中的每个element都有唯一的Key，Key可以用来检索元素。(Key是static静态变量，全局唯一，为Element提供唯一性保障)
//这里在模拟CoroutineContext
//                     ----->  CombinedContext                                       ------>CoroutineExceptionHandler
//                                       -------->Job                                ------>CoroutineDispatcher
//  CoroutineContext   ----->  Element   -------->AbstractCoroutineContextElement    ------>CoroutineName
//                     ----->  EmptyCoroutineContext
//CoroutineContext.get方法
//  在CoroutineContext中为抽象方法：
//  Element中的实现：只要Key与当前Element的Key匹配上了，返回该Element否则返回null。
//  CombinedContext中的实现：遍历链表，查询与Key相等的Element，如果没找到返回null。
//  EmptyCoroutineContext：返回null
//CoroutineContext.plus方法
//  两个CoroutineContext的子类相加， 我们令其形式为coroutineContext.plus(context)
//  情况一：context为EmptyCoroutineContext，返回coroutineContext
//  情况二：coroutineContext与context为相同的KEY，返回context
//  情况三：coroutineContext中没有ContinuationInterceptor，返回(去掉了context的coroutineContext,context)
//  情况四：coroutineContext中有ContinuationInterceptor,去掉context和ContinuationInterceptor后coroutineContext变成了EmptyCoroutineContext，返回CombinedContext(context,interceptor)(这个interceptor是coroutineContext中的ContinuationInterceptor)
//  情况五：coroutineContext中有ContinuationInterceptor,去掉context和ContinuationInterceptor后coroutineContext中还有别的Element，返回CombinedContext(CombinedContext(去掉context和ContinuationInterceptor后的coroutineContext, context), interceptor)
//CoroutineContext.minusKey方法
//  EmptyCoroutineContext：返回本身
//  Element minusKey方法：如果Key与当前element的Key相等，返回EmptyCoroutineContext，否则相当于没减成功，返回当前element
//  CombinedContext minusKey方法：删除链表中符合条件的节点，分三种情况
//  节点不在尾部：从右到左，去掉第一个对应KEY的Element。
//  节点在尾部：去掉尾部节点
//  没找到节点：原样返回
//  三种情况在代码中处理方式不一样，但都是减去对应KEY的Element

//协程的结构化并发
//



//协程启动模式
//CoroutineStart.DEFAULT
//立即执行
//CoroutineStart.LAZY
//有需要才会运行
//CoroutineStart.ATOMIC
//使用后 协程在 遇到第一个 挂起点时 才会取消
//Coroutine.UNDISPATCHED
//不进行任何调度,包括线程切换等, 线程状态会跟随父协程保持一致,直到遇见第一个suspend函数

//取消
//对于delay()来说，它可以自动检测当前的协程是否已经被取消，如果已经取消的话，抛出CancellationException
//
//
//
//


//异常
//SupervisorJob()与Job()
//Job的子协程发生异常被取消会同时取消Job的其它子协程，而SupervisorJob不会。
//例子
fun testForSupervisorJob() {
    runBlocking {
        val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
            printlnThread("catch 到了 $throwable")
        }

        val customScope =
            CoroutineScope(SupervisorJob() + CoroutineName("自定义协程") + Dispatchers.IO + exceptionHandler)

        customScope.launch {
            printlnThread("准备出错1")
            throw KotlinNullPointerException("错误一")
        }.join()

        customScope.launch {
            printlnThread("准备出错2")
            throw KotlinNullPointerException("错误二")
        }.join()

    }
}
//
//


//Flow

//onStart（数据发射之前触发，onStart所在的线程，是数据产生的线程）
//onCompletion（数据流结束时触发，onCompletion所在的线程，是数据产生的线程）
//onEmpty（数据流结束了，却没有发出任何元素的时候触发）
//flowOn（指定上游数据流的CoroutineContext，下游数据流不会受到影响）
//catch（数据流发生错误的时候触发）
fun testFlow01(scope: CoroutineScope) {

    val flow1 = flowOf(1, 2, 3)

    val flow2 = flowOf<Int>()

    val start = System.currentTimeMillis()

    scope.launch(Dispatchers.Default) {

        flow1
            .onStart {
                LogUtil.default01("Flow开始")
            }
            .onEach {
                delay(100)
                LogUtil.default01("emit ${Thread.currentThread().name}")
                LogUtil.default01(it.toString())
            }
            .flowOn(Dispatchers.Main)
            .onCompletion {
                LogUtil.default01("Flow完成")
            }
            .onEmpty {
                LogUtil.default01("Flow空的")
            }
            .collect {
                LogUtil.default01("collect $it ${Thread.currentThread().name}")
                delay(100)
            }


        LogUtil.default01("时间  ${(System.currentTimeMillis() - start)}")
    }

    scope.launch(Dispatchers.Default) {

        flow2
            .onEmpty {
                LogUtil.default01("这是空的")
                throw Exception("错误")
            }
            .catch {
                LogUtil.default01("抓到错误了")
            }
            .collect {
                LogUtil.default01(it.toString())
            }
    }
}

//transform（对发射的值进行变换）
//transformLatest（当有新值发送时如果上个变换还没结束，会先取消掉）
//transformWhile（返回值是Boolean,如果为False则不再进行后续变换,为True则继续执行）
@OptIn(ExperimentalCoroutinesApi::class)
fun testFlow02(scope: CoroutineScope) {
    scope.launch(Dispatchers.Main) {
        flow {

            delay(100)
            emit(1)

            delay(100)
            emit(2)

            delay(100)
            emit(3)

            delay(100)
            emit(4)

            delay(100)
            emit(5)

            delay(100)
            emit(6)

            delay(100)
            emit(7)
        }
            .transformWhile {
                emit(it + 1)
                it != 4
            }
            .transform {
                emit(it + 1)
            }
            .transformLatest {
                LogUtil.default01("transform1 开始")
                delay(200)
                emit(it + 1)
                LogUtil.default01("transform1 结束")
            }
            .onEach {
                LogUtil.default01("each01 $it")
            }
            .launchIn(CoroutineScope(Dispatchers.IO))

    }
}

//flow操作符
//map
//filter
//onEach（用来遍历每一条数据的）
//debounce（确保flow的各项数据之间存在一定的时间间隔，如果是时间点过于临近的数据只会保留最后一条）
@OptIn(FlowPreview::class)
fun test01(scope: CoroutineScope) {

    scope.launch {

        val flow = flow {
            emit(0)
            delay(300)
            emit(1)
            delay(200)
            emit(2)
            delay(300)
            emit(3)
            delay(500)
            emit(4)
            delay(600)
            emit(5)
            delay(100)
            emit(6)
        }

        flow
            .filter {
                it != 2
            }
            .map {
                it * it
            }
            .onEach {
                LogUtil.default01("流中的数据$it")
            }
            .debounce(500)
            .collect {
                LogUtil.default01(it.toString())
            }
    }

}

//reduce（通过参数给我们一个Flow的累积值和一个Flow的当前值，我们可以在函数体中对它们进行一定的运算，运算的结果会作为下一个累积值继续传递到reduce函数当中）
fun test02(scope: CoroutineScope) {

    scope.launch {

        val temp = flow {
            emit("Hello")
            emit("World")
            emit("!")
        }.reduce { accumulator, value ->
            accumulator + value
        }

        LogUtil.default01(temp)

    }
}

//fold（fold和reduce主要区别在于，fold函数需要传入一个初始值，这个初始值会作为首个累积值被传递到fold的函数体当中）
fun test03(scope: CoroutineScope) {

    scope.launch {

        val temp = flow {
            emit("Hello")
            emit("World")
            emit("!")
        }
            .fold("WZY,") { acc: String, value: String ->
                acc + value
            }

        LogUtil.default01(temp)

    }

}

//flatMapConcat
@OptIn(FlowPreview::class)
fun test04(scope: CoroutineScope) {
    scope.launch {

        flowOf(1, 2, 3)
            .flatMapConcat {
                listOf("A$it", "B$it").asFlow()
            }
            .collect {
                LogUtil.default01(it)
            }
    }

}

//flatMapMerge（与flatMapConcat的区别：concat是连接的意思，merge是合并的意思。连接一定会保证数据是按照原有的顺序连接起来的，而合并则只保证将数据合并到一起，并不会保证顺序）
@OptIn(FlowPreview::class)
fun test05(scope: CoroutineScope) {

    scope.launch {

        val time = System.currentTimeMillis()

        flowOf(300, 200, 100)
            .flatMapMerge {

                flow {
                    delay(it.toLong())
                    emit("A$it")
                    emit("B$it")
                }

            }.collect {
                LogUtil.default01(it)
            }

        LogUtil.default01((System.currentTimeMillis() - time).toString())

    }

}

//collectLatest（只接收处理最新的数据，如果有新数据到来了而前一个数据还没有处理完，则会将前一个数据剩余的处理逻辑全部取消）
fun test06(scope: CoroutineScope) {

    scope.launch {

        flow {
            emit(1)
            delay(100)
            emit(2)
        }.collectLatest {

            LogUtil.default01("开始收集 $it")
            delay(300)
            LogUtil.default01("结束收集 $it")

        }

    }

}

//flatMapLatest（flow1中的数据传递到flow2中会立刻进行处理，但如果flow1中的下一个数据要发送了，而flow2中上一个数据还没处理完，则会直接将剩余逻辑取消掉，开始处理最新的数据）
@OptIn(ExperimentalCoroutinesApi::class)
fun test07(scope: CoroutineScope) {

    scope.launch {
        flow {
            emit(1)
            delay(100)
            emit(2)
            delay(200)
            emit(3)
        }.flatMapLatest {
            flow {
                emit("A $it")
                delay(50)
                emit("B $it")
                delay(300)
                emit("C $it")
            }
        }.collect {
            LogUtil.default01(it)
        }
    }

}

//zip
fun test08(scope: CoroutineScope) {

    scope.launch {

        val start = System.currentTimeMillis()

        val flow1 = flow {

            emit("a")
            delay(100)

            emit("b")
            delay(300)

            emit("c")
            delay(100)
        }

        val flow2 = flow {

            emit(1)
            delay(200)

            emit(2)
            delay(100)

            emit(4)
            delay(100)

            emit(5)
            delay(300)

            emit(6)
            delay(100)
        }



        flow1.zip(flow2) { a, b ->
            "$a  $b"
        }.collect {
            LogUtil.default01((System.currentTimeMillis() - start).toString())
            LogUtil.default01(it)
        }

    }

}

//buffer（让flow函数和collect函数运行在不同的协程当中，这样flow中的数据发送就不会受collect函数的影响了）
fun test09(scope: CoroutineScope) {

    scope.launch {

        val start = System.currentTimeMillis()

        flow {
            emit(1)
            delay(1000)
            emit(2)
            delay(1000)
            emit(3)
        }.onEach {
            LogUtil.default01("$it is ready")
        }.buffer()
            .collect {
                delay(1000)
                LogUtil.default01((System.currentTimeMillis() - start).toString())
                LogUtil.default01("$it is handled")
            }
    }
}

//conflate
fun test10(scope: CoroutineScope) {

    scope.launch {

        flow {
            emit(1)
            delay(100)
            emit(2)
            emit(3)
            emit(4)
            delay(100)
            emit(5)
            delay(100)
            emit(6)
            delay(200)
            emit(7)
        }
            .conflate()
            .collect {
                LogUtil.default01("start handle $it")
                delay(400)
                LogUtil.default01("finish handle $it")
            }
    }
}

//SharedFlow
//参数
//replay（重放数据个数）
//当新订阅者时注册时会重放缓存的 replay 个数据
//extraBufferCapacity（额外缓存容量）
//额外缓存容量，在 replay 之外的额外容量，SharedFlow 的缓存容量 capacity = replay + extraBufferCapacity
//onBufferOverflow（缓存溢出策略）
//缓存容量 capacity 满时的处理策略（SUSPEND、DROP_OLDEST、DROP_LAST）
//默认是BufferOverflow.SUSPEND，当额外缓冲区满后，挂起emit函数，暂停发送数据。
fun testFlow03(scope: CoroutineScope) {

    val sharedFlow = MutableSharedFlow<Int>(1, 3, BufferOverflow.DROP_OLDEST)

    sharedFlow.onEach {
        LogUtil.default01("1号观察者： 数据$it")
    }.launchIn(scope)

    scope.launch {
        for (i in 1..5) {
            sharedFlow.emit(i)
            delay(100)
        }
    }

    sharedFlow.onEach {
        LogUtil.default01("2号观察者： 数据$it")
    }.launchIn(scope)

}


//Channel
fun testChannel01(scope: CoroutineScope) {

    val channel = Channel<Int>()

    scope.launch {
        (1..4).forEach {
            LogUtil.default01("Send $it")
            channel.send(it)
        }
    }

    scope.launch {
        for (i in channel) {
            LogUtil.default01("Receive: $i")
        }
    }
}

@OptIn(ExperimentalCoroutinesApi::class)
fun testChannel02(scope: CoroutineScope) {

    scope.launch {

        val channel: ReceiveChannel<Int> = produce {
            (1..3).forEach {
                send(it)
                LogUtil.default01("send $it")
            }
        }

        launch {
            channel.consumeEach {
                LogUtil.default01("receive  $it")
            }
        }

    }


}

@OptIn(ExperimentalCoroutinesApi::class)
fun testChannel03(scope: CoroutineScope) {

    val channel = Channel<Int>(capacity = 3)

    scope.launch {
        (1..3).forEach {
            channel.send(it)
            LogUtil.default01("Send $it")
        }

        channel.close()

        LogUtil.default01("AisClosedForSend  ${channel.isClosedForSend}")
        LogUtil.default01("AisClosedForReceive  ${channel.isClosedForReceive}")
    }

    scope.launch {
        for (element in channel) {
            LogUtil.default01("receive $element")
            delay(1000)

            LogUtil.default01("BisClosedForSend  ${channel.isClosedForSend}")
            LogUtil.default01("BisClosedForReceive  ${channel.isClosedForReceive}")
        }

        LogUtil.default01("CisClosedForSend  ${channel.isClosedForSend}")
        LogUtil.default01("CisClosedForReceive  ${channel.isClosedForReceive}")
    }

}

@OptIn(ObsoleteCoroutinesApi::class)
fun testChannel04(scope: CoroutineScope) {
    //BroadcastChannel多个接收端不存在互斥行为

    val channel = BroadcastChannel<Int>(3)

    scope.launch {
        (1..3).forEach {
            delay(100)
            channel.send(it)
        }
    }

    scope.launch {
        val receiveChannel = channel.openSubscription()
        for (i in receiveChannel) {
            LogUtil.default01("A receive $i")
        }
    }


    scope.launch {
        val receiveChannel = channel.openSubscription()
        for (i in receiveChannel) {
            LogUtil.default01("B receive $i")
        }
    }

    scope.launch {
        val receiveChannel = channel.openSubscription()
        for (i in receiveChannel) {
            LogUtil.default01("C receive $i")
        }
    }


}


//Select
private suspend fun getCacheInfo01(id: String): Product {
    delay(100)
    return Product(id, 9.88)
}

private suspend fun getCacheInfo02(id: String): Product {
    delay(98)
    return Product(id, 9.66)
}

private suspend fun getNetWork(id: String): Product {
    delay(200)
    return Product(id, 7.88)
}

data class Product(val id: String, val price: Double, var isCache: Boolean = false)

fun testSelect01(scope: CoroutineScope) {

    val id = "1122"

    val cacheDeferred01 = scope.async {
        getCacheInfo01(id)
    }
    val cacheDeferred02 = scope.async {
        getCacheInfo02(id)
    }
    val latestDeferred01 = scope.async {
        getNetWork(id)
    }
    val product = scope.async {
        select {

            cacheDeferred01.onAwait {
                it.copy(isCache = true)
            }

            cacheDeferred02.onAwait {
                it.copy(isCache = true)
            }

            latestDeferred01.onAwait {
                it.copy(isCache = false)
            }
        }
    }

    scope.launch {
        if (product.await().isCache) {
            LogUtil.default01(latestDeferred01.await().toString())
        }
    }
}

@OptIn(ExperimentalCoroutinesApi::class)
fun testSelect02(scope: CoroutineScope) {

    scope.launch {

        val channel1 = produce {
            delay(100)
            send("1")
            delay(500)
        }
        val channel2 = produce {
            delay(100)
            send("a")
            delay(200)
            send("b")
            delay(200)
            send("c")
        }
        repeat(6) {
            val result = select {
                channel1.onReceiveCatching {
                    LogUtil.default01(it.getOrNull() ?: "Channel已经关闭")
                }

                channel2.onReceiveCatching {
                    it.also {
                        LogUtil.default01(it.getOrNull() ?: "Channel已经关闭")
                    }
                }
            }

            LogUtil.default01("result $result")
        }

        channel1.cancel()
        channel2.cancel()
    }
}

private suspend fun <T> fastest(vararg defers: Deferred<T>): T = select {

    fun cancelAll() = defers.forEach {
        it.cancel()
    }
    for (deferred in defers) {
        deferred.onAwait {
            cancelAll()
            it
        }
    }
}

fun testSelect03(scope: CoroutineScope) {

    scope.launch {

        val deferred1 = async {
            delay(100)
            LogUtil.default01("done 1")
            "done 1"
        }
        val deferred2 = async {
            delay(50)
            LogUtil.default01("done 2")
            "done 2"
        }
        val deferred3 = async {
            delay(1000)
            LogUtil.default01("done 3")
            "done 3"
        }
        val deferred4 = async {
            delay(2000)
            LogUtil.default01("done 4")
            "done 4"
        }
        val deferred5 = async {
            delay(700)
            LogUtil.default01("done 5")
            "done 5"
        }


        val result = fastest(deferred1, deferred2, deferred3, deferred4, deferred5)

        LogUtil.default01("已完成 $result")

    }

}


